package org.apache.solr.ocean.replication;

import java.io.File;
import java.io.IOException;
import java.text.DecimalFormat;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeSet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.solr.ocean.replication.LogFile2.Record;
import org.apache.solr.ocean.replication.RawLogFile.RecordHeader;
import org.apache.solr.ocean.replication.RawLogFile.StreamRecord;
import org.apache.solr.ocean.util.ConcurrentLinkedHashMap;
import org.apache.solr.ocean.util.ConcurrentTreeMap;

public class LogFileManager {
  public static Logger log = Logger.getLogger(LogFileManager.class.getName());
  public static final long MAX_FILE_SIZE = 1024 * 1024 * 64;
  private File directoryFile;
  private ConcurrentLinkedHashMap<Long,Record> recordLruMap;
  private ConcurrentTreeMap<Long,LogFile2> logFiles = new ConcurrentTreeMap<Long,LogFile2>();
  private AtomicLong sequence = new AtomicLong(0);
  private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  private ScheduledExecutorService logFileCheckTimer;

  public LogFileManager(File directoryFile) throws IOException {
    this.directoryFile = directoryFile;
    recordLruMap = new ConcurrentLinkedHashMap<Long,Record>(200);
    File[] files = directoryFile.listFiles();
    for (File file : files) {
      if (!file.isDirectory()) {
        long id = getLogFileNumber(file.getName());
        LogFile2 logFile = new LogFile2(id, file);
        logFiles.put(id, logFile);
      }
    }
    if (logFiles.size() > 0) {
      long last = logFiles.lastKey();
      long next = last + 1;
      sequence.set(next);
    }
    logFileCheckTimer = Executors.newSingleThreadScheduledExecutor();
    logFileCheckTimer.scheduleWithFixedDelay(new LogFileSizeCheck(), 1000, 10 * 1000, TimeUnit.MILLISECONDS);
  }

  /**
   * public byte[] getCheckSum(String category, Long lowId, Long highId) throws
   * Exception { MessageDigest messageDigest = MessageDigest.getInstance("MD5");
   * for (LogFile2 logFile : logFiles.values()) { if
   * (!logFile.checksum(category, lowId, highId, messageDigest)) { break; } }
   * return messageDigest.digest(); }
   * 
   * public byte[] getCheckSum(Long lowId, Long highId) throws Exception {
   * MessageDigest messageDigest = MessageDigest.getInstance("MD5"); for
   * (LogFile2 logFile : logFiles.values()) { if (!logFile.checksum(lowId,
   * highId, messageDigest)) { break; } } return messageDigest.digest(); }
   */
  public class LogFileSizeCheck implements Runnable {
    public void run() {
      try {
        LogFile2 logFile = logFiles.lastValue();
        if (logFile.size() >= MAX_FILE_SIZE) {
          LogFile2 newLogFile = createNewLogFile();
          logFiles.put(newLogFile.getId(), newLogFile);
        }
      } catch (IOException ioException) {
        log.log(Level.SEVERE, "", ioException);
      }
    }
  }

  public boolean delete(Long id) throws Exception {
    LogFile2 logFile = getLogFileContaining(id);
    if (logFile == null)
      throw new Exception("unknown id: " + id);
    return logFile.delete(id);
  }

  public Long getPreviousId(Long id) {
    LogFile2 logFile = getLogFileContaining(id);
    Long previousId = logFile.getPreviousId(id);
    if (previousId == null) {
      SortedMap<Long,LogFile2> subMap = logFiles.headMap(logFile.getId());
      Long lastKey = subMap.lastKey();
      if (lastKey != null) {
        logFile = subMap.get(lastKey);
        previousId = logFile.getPreviousId(id);
      }
    }
    return previousId;
  }

  public Long getHighestId(String category) {
    List<Long> list = new LinkedList<Long>();
    for (LogFile2 logFile : logFiles.values()) {
      list.add(logFile.getHighestID(category));
    }
    return Collections.max(list);
  }

  public Long getLowestId() {
    TreeSet<Long> sortedSet = new TreeSet<Long>();
    for (LogFile2 logFile : logFiles.values()) {
      sortedSet.add(logFile.getLowestId());
    }
    return sortedSet.first();
  }

  public Long getHighestId() {
    TreeSet<Long> sortedSet = new TreeSet<Long>();
    for (LogFile2 logFile : logFiles.values()) {
      sortedSet.add(logFile.getHighestId());
    }
    return sortedSet.last();
  }

  public LogFile2 createNewLogFile() throws IOException {
    long id = sequence.getAndIncrement();
    String fileName = createLogFileName(id);
    File file = new File(directoryFile, fileName);
    LogFile2 logFile = new LogFile2(id, file);
    lock.writeLock().lock();
    try {
      logFiles.put(id, logFile);
      return logFile;
    } finally {
      lock.writeLock().unlock();
    }
  }

  private static String createLogFileName(long value) {
    DecimalFormat decimalFormat = (DecimalFormat) DecimalFormat.getInstance();
    decimalFormat.applyPattern("00000000");
    return "log" + decimalFormat.format(value) + ".bin";
  }

  private static long getLogFileNumber(String fileName) {
    String numberString = fileName.substring(3, fileName.lastIndexOf('.'));
    return Long.parseLong(numberString);
  }

  private LogFile2 getLogFileContaining(Long id) {
    lock.readLock().lock();
    try {
      for (LogFile2 logFile : logFiles.values()) {
        if (logFile.containsId(id)) {
          return logFile;
        }
      }
      return null;
    } finally {
      lock.readLock().unlock();
    }
  }

  public boolean contains(Long id) {
    lock.readLock().lock();
    try {
      for (LogFile2 logFile : logFiles.values()) {
        if (logFile.containsId(id)) {
          return true;
        }
      }
      return false;
    } finally {
      lock.readLock().unlock();
    }
  }

  // TODO: doesn't return greater than
  private LogFile2 getLogFileContaining(String category, Long id) {
    lock.readLock().lock();
    try {
      for (LogFile2 logFile : logFiles.values()) {
        if (logFile.containsId(category, id)) {
          return logFile;
        }
      }
      return null;
    } finally {
      lock.readLock().unlock();
    }
  }
  /**
   * public List<RecordHeader> getRecordHeaders(Long snapshotId, int num)
   * throws Exception { LogFile2 logFile = getLogFileContaining(snapshotId); if
   * (logFile == null) throw new Exception("unknown snapshotid: " + snapshotId); //
   * TODO: handle getting more if needed from next log file return
   * logFile.getRecordHeaders(snapshotId, num); }
   */
  
  public RecordIterator getRecordIterator(Long snapshotId) throws IOException {
    return new RecordIterator(snapshotId);
  }
  
  public RecordIterator getRecordIterator(String category, Long snapshotId) throws Exception {
    
  }
  
  public class RecordIterator {
    private Iterator<LogFile2> logFileIterator;
    private LogFile2.RecordIterator currentRecordIterator;
    
    public RecordIterator(Long snapshotId) throws IOException {
      Iterator<LogFile2> logFileIterator = logFiles.values().iterator();
      while (logFileIterator.hasNext()) {
        LogFile2 logFile = logFileIterator.next();
        if (logFile.containsId(snapshotId)) {
          currentRecordIterator = logFile.getRecordIterator(snapshotId);
        }
      }
    }
    
    public void close() throws IOException {
      if (currentRecordIterator != null) currentRecordIterator.close();
    }
    
    public Record next() throws IOException {
      if (currentRecordIterator != null) {
        return currentRecordIterator.next();
      } else {
        return null;
      }
    }
    
    public boolean hasNext() throws IOException {
      if (currentRecordIterator == null) return false;
      if (currentRecordIterator.hasNext()) {
        return true;
      } else {
        currentRecordIterator.close();
        if (logFileIterator.hasNext()) {
          LogFile2 logFile = logFileIterator.next();
          currentRecordIterator = logFile.getRecordIterator(null);
          return hasNext();
        } 
      }
      return false;
    }
  }
  /**
  public Record getRecordFromId(Long snapshotId) throws IOException {
    LogFile2 logFile = getLogFileContaining(snapshotId);
    if (logFile == null)
      return null;
    return logFile.getRecordFromId(snapshotId);
  }

  public Record getRecordFromId(String category, Long snapshotId) throws IOException {
    LogFile2 logFile = getLogFileContaining(category, snapshotId);
    if (logFile == null)
      return null;
    return logFile.getRecordGreaterOrEqualToID(category, snapshotId);
  }

  public Record getRecord(RecordHeader recordHeader) throws Exception {
    LogFile2 logFile = getLogFileContaining(recordHeader.id);
    if (logFile == null)
      throw new Exception("unknown snapshotid: " + recordHeader.id);
    // TODO: handle getting more if needed from next log file
    LogFile2.Reader reader = logFile.getReader();
    return reader.getRecord(recordHeader);
  }
  **/
  public void commit(RecordHeader recordHeader) {
    // TODO: implement commit
  }

  public RecordHeader writeRecord(Long id, String category, float schemaVersion, StreamRecord streamRecord) throws IOException {
    LogFile2 logFile = getCurrentLogFile();
    return logFile.writeRecord(id, category, schemaVersion, streamRecord);
  }

  /**
   * public Record getRecordGreaterOrEqualTo(Long snapshotId) throws IOException {
   * LogFile2 logFile = getLogFileContaining(snapshotId); if (logFile == null)
   * return null; return logFile.getRecordFromId(snapshotId); }
   * 
   * public Record getRecordGreaterOrEqualTo(EnumSet<Load> load, String
   * category, Long snapshotID, ByteBufferPool documentsByteBuffers,
   * ByteBufferPool deletesByteBuffers) throws IOException { LogFile2 logFile =
   * getLogFileContaining(category, snapshotID); if (logFile == null) return
   * null; return logFile.getRecordGreaterOrEqualToID(load, category,
   * snapshotID, documentsByteBuffers, deletesByteBuffers); }
   */
  public LogFile2 getCurrentLogFile() throws IOException {
    LogFile2 logFile = logFiles.lastValue();
    if (logFile == null) {
      logFile = createNewLogFile();
    }
    return logFile;
  }
}
